package com.exsun.gateway.jt809.kafka;

import com.exsun.gateway.jt809.codec.protocol.Jt809Decode;
import com.exsun.gateway.jt809.common.DevInfoService;
import com.exsun.gateway.jt809.config.Jt809Config;
import com.exsun.gateway.jt809.data.PushData;
import com.exsun.gateway.jt809.handler.HttpRequest;
import com.exsun.gateway.jt809.handler.ResponseHandler;
import com.exsun.gateway.jt809.handler.cache.MediaListHandle;
import com.exsun.gateway.jt809.kafka.gps.*;
import com.exsun.gateway.jt809.kafka.gps.check.GpsReasonableHandle;
import com.google.gson.reflect.TypeToken;
import exsun.jt808.data.base.StatusResult;
import exsun.jt808.data.config.jt808.Jt808DistributeMsgTopic;
import exsun.jt808.data.config.jt808.Jt808ReceiveMsgTopic;
import exsun.jt808.data.config.jt809.Jt809MsgIdConfig;
import exsun.jt808.data.distribute_data.jt1078.QueryMediaList;
import exsun.jt808.data.jt809_data.BasicVehicleInfo;
import exsun.jt808.data.jt809_data.SendBaseEntity;
import exsun.jt808.data.jt809_data.video.main.UpAlarmMediaList;
import exsun.jt808.data.jt809_data.video.slave.DownRealVideoFileListReq;
import exsun.jt808.data.kafka_data.KafkaJT808Data;
import exsun.jt808.data.upload_data.GpsData;
import exsun.jt808.data.util.GsonUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import java.lang.reflect.Type;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

@Service
public class JT809KafkaGpsDataHandler extends JT809KafkaHandleBase<GpsData> {
    private Logger logger = LoggerFactory.getLogger(JT809KafkaGpsDataHandler.class);

    @Autowired
    private DevInfoService devInfoService;

    @Autowired
    private GpsHandle gpsHandle;

    @Autowired
    Jt809Decode jt809Decode;

    @Autowired
    ResponseHandler responseHandler;

    @Autowired
    Jt809Config jt809Config;

    private static final int TIMEOUT = 3000;

    //订阅GPS数据
    @KafkaListener(topics = Jt808ReceiveMsgTopic.cleanGpsInfo, containerFactory = "gpsData809Factory")
    public void batchGPSMsgConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        try {

            Type type = new TypeToken<GpsData>() {
            }.getType();
            List<PushData> pushs = super.done1(records, type, (obj) -> {
                return handle(obj);
            });
            //发送错误，则不提交偏移量
            if (pushs == null) return;
            if (pushs.size() != 0) {
                logger.info("上报报警：{}条,上报gps:{}条",
                        pushs.stream().filter(v -> v.getMsgId() == Jt809MsgIdConfig.UP_WARN_MSG).count(),
                        pushs.stream().filter(v -> v.getMsgId() == Jt809MsgIdConfig.UP_EXG_MSG).count());
            }
            ack.acknowledge();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

//    @KafkaListener(topics = Jt808ReceiveMsgTopic.gpsData, containerFactory = "gpsData809Factory")
//    public void batchGPSMsgConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
//        try {
//            Type type = new TypeToken<KafkaJT808Data<GpsData>>() {}.getType();
//            List<PushData> pushs = super.done(records, type, this::handle);
//            //发送错误，则不提交偏移量
//            if (pushs == null) return;
//            if (pushs.size() != 0) {
//                logger.info("上报报警：{}条,上报gps:{}条",
//                        pushs.stream().filter(v -> v.getMsgId() == Jt809MsgIdConfig.UP_WARN_MSG).count(),
//                        pushs.stream().filter(v -> v.getMsgId() == Jt809MsgIdConfig.UP_EXG_MSG).count()
//                );
//            }
//            ack.acknowledge();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }

    private List<PushData> handle(GpsData gpsData) {
        List<PushData> pushs = new ArrayList<>();
//        //测试
//        if(gpsData.gte + 60000 < new Date().getTime()){
//            return pushs;
//        }

        if(gpsData.gte + 20*3600000 < new Date().getTime()){
            return pushs;
        }

        BasicVehicleInfo basicVehicleInfo = devInfoService.getBasicVehicleInfo(gpsData.dvo);
        if (basicVehicleInfo == null) return pushs;

        if (gpsData.amm != 0)
        {
            httpPostToGetMediaList(gpsData);
        }
        //todo 判断历史数据
        boolean bHistory = GpsReasonableHandle.gpsIsHistory(gpsData);
        if (bHistory) {
            //历史数据采用补传方式上传
            List<PushData> historyLocation = gpsHandle.getPushData(new HistoryLocation(), gpsData, basicVehicleInfo);
            pushs.addAll(historyLocation);
            return pushs;
        }

        GpsReasonableHandle.addGpsCache(gpsData);
        //实时数据上传：

//        // 报警数据上传
//        List<PushData> alarmInfo = gpsHandle.getPushData(new AlarmInfo(), gpsData, basicVehicleInfo);
//        pushs.addAll(alarmInfo);
        // 实时定位信息上传
        List<PushData> realLocation = gpsHandle.getPushData(new RealLocation(), gpsData, basicVehicleInfo);
        pushs.addAll(realLocation);
        //上报报警信息消息
        List<PushData> upWarnMsgInfoEx = gpsHandle.getPushData(new WarnMsgInfoEx(), gpsData, basicVehicleInfo);
        pushs.addAll(upWarnMsgInfoEx);


        return pushs;
    }

    private void httpPostToGetMediaList(GpsData gpsData){
        Long now = gpsData.gte;
        Long start = now - 5 * 60 * 1000;
        Long end = now + 5 * 60 * 1000;
        Date startTime = new Date(start);
        Date endTime = new Date(end);
        List<Integer> channels = responseHandler.getAllChannelIds(gpsData.dvo);
        QueryMediaList entity = new QueryMediaList();
        for (Integer channel : channels) {
            entity.setChanelId(channel);
        }
        entity.setBeginTime(startTime);
        entity.setEndTime(endTime);
        entity.setGpsAlarmFlag((int) gpsData.amm * 1000000);
        entity.setMediaAlarmFlag((int) gpsData.vam * 1000000);
        entity.setMediaType(0);
        entity.setCodeType(0);
        entity.setStorageType(0);
        entity.response = false;
        entity.devNo = gpsData.dvo;
        entity.requestId = "3412568";
        entity.msgId = (int) Long.parseLong("9205", 16);
        entity.serialNo = 123456;
        entity.userId = "123456";
        entity.userName = "java809";
        entity.distributeTime = new Date();
        entity.distributeMsgTopic = Jt808DistributeMsgTopic.QueryMediaList_Send;
        HttpRequest.postOrder(jt809Config.getHttpRequestUrl(), entity, TIMEOUT);
        logger.info("告警后请求列表");
    }
}
